package com.atguigu.bigdata.spark.streaming;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;


import java.util.*;

public class SparkStreaming10_MockData_JAVA {
    public static void main(String[] args) throws InterruptedException {
        // 生成模拟数据
        // 格式 ：timestamp area city userid adid
        // 含义： 时间戳   区域  城市 用户 广告

        // Application => Kafka => SparkStreaming => Analysis

        Map<String, Object> kafkaPara = new HashMap<>();
        kafkaPara.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        kafkaPara.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        kafkaPara.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        String topic = "test111";

        KafkaProducer<String,String> producer = new KafkaProducer<>(kafkaPara);
        while(true) {
            List<String> rawData = mockData();
            for(String val : rawData) {
                ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, val);
                producer.send(record);
                System.out.println(record);
            }
            Thread.sleep(2000);
        }
    }
    private static List<String> mockData() {
        List<String> areaList = Arrays.asList("华北", "华东", "华南");
        List<String> cityList = Arrays.asList("北京", "上海", "深圳");
        List<String> list = new ArrayList<>();

        for(int i = 0; i <= new Random().nextInt(50); i++) {
            String area = areaList.get(new Random().nextInt(3));
            String city =cityList.get(new Random().nextInt(3));
            int userid = new Random().nextInt(6) + 1;
            int adid = new Random().nextInt(6) + 1;
            String value = System.currentTimeMillis() + " " + area + " " + city + " " +  userid + " " + adid;
            list.add(value);
        }
        return list;
    }
}
